package org.medusa.network.tcp;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;

import org.medusa.buffer.RingBuffer;
import org.medusa.core.Core;
import org.medusa.network.Connector;
import org.medusa.packet.blob.BasePacket;

//1.Sync与Async Connector互相转换的问题
//2.连接超时设置
public abstract class BaseTcpSyncConnector<PacketType extends BasePacket> extends
		Connector<PacketType> {
	private Socket fd = null;
	private String host = "localhost";
	private int port = 0;
	private int conn_timeout_ms;

	protected RingBuffer ringbuf = null;
	private byte[] rcvbuf = null;
	private int recv_buf_size = 0;
	private int send_buf_size = 0;
	// private AtomicBoolean running;
	// other
	private BaseTcpSyncConnector<PacketType> self;
	private InputStream input = null;
	private OutputStream output = null;

	public BaseTcpSyncConnector(int rcv_buf, int snd_buf) {
		this.recv_buf_size = rcv_buf;
		this.send_buf_size = snd_buf;

		this.ringbuf = new RingBuffer(this.recv_buf_size, this.recv_buf_size/2);
		this.rcvbuf = new byte[this.recv_buf_size / 5];

		// this.running = new AtomicBoolean(false);
		// 其他
		this.self = this;
	}

	@Override
	public Error Connect(String ip, int port, int timeout_ms) {
		if (this.is_running()) {
			return new Error("already connect (" + this.host + ":" + this.port
					+ ")");
		}
		this.host = ip;
		this.port = port;
		this.conn_timeout_ms = timeout_ms;

		try {
			this.fd = new Socket(/* ip, port */);
			this.fd.connect(new InetSocketAddress(ip, port), timeout_ms);

			this.fd.setReceiveBufferSize(this.recv_buf_size);
			this.fd.setSendBufferSize(this.send_buf_size);
			this.input = this.fd.getInputStream();
			this.output = this.fd.getOutputStream();
			this.set_state_running();
			// this.Success("ConnectionMade: (%v -> %v) err:%v",
			// self.GetLocalAddr(), self.GetRemoteAddr(), err);
		} catch (UnknownHostException e) {
			// e.printStackTrace();
			this.release();
			return new Error(e.getMessage());

		} catch (IOException e) {
			// e.printStackTrace();
			// this.release();
			return new Error(e.getMessage());
		}
		return null;
	}

	@Override
	public void Reconnect(int sleep_ms, String tip) {
		Error err = this.Connect(this.host, this.port, this.conn_timeout_ms);
		while (err != null) {
			Core.Sleep(sleep_ms);
			//this.Warning(tip);
			err = this.Connect(this.host, this.port, this.conn_timeout_ms);
		}
	}

	@Override
	public void Close() {
		this.release();
	}

	@Override
	public boolean IsConnected() {
		return this.is_running() ? (this.fd.isConnected() && !this.fd
				.isClosed()) : false;
	}

	@Override
	public void SendPacket(PacketType outpack) {
		Error err = null;
		if (self.fd != null && this.is_running()) {
			err = outpack.Write2IO(self.output);
			if (err != null) {
				self.release();
			}
		}
	}

	public PacketType RecvPacket() {
		PacketType inpack = null;
		int received;

		while (this.is_running()) {
			try {
				received = self.input.read(self.rcvbuf);
				if (received <= 0) {
					self.release();
					break;
				}
				self.ringbuf.WriteBlob(self.rcvbuf, received);
				inpack = self.parse_packet();
				if (inpack != null)
					break;
				// runtime.Gosched(); continue;
			} catch (IOException e) {
				// e.printStackTrace();
				self.release();
				break;
			}

		}// end of : while
		return inpack;
	}

	public synchronized PacketType SendCommand(PacketType outpack) {
		PacketType recv_pack = null;

		this.SendPacket(outpack);
		if (this.is_running())
			recv_pack = this.RecvPacket();
		return recv_pack;
	}

	private void release() {
		this.set_state_stopped();
		if (self.fd != null) {
			try {
				this.input.close();
				this.output.close();
				this.fd.close();

				this.input = null;
				this.output = null;

				this.fd = null;

			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	protected abstract PacketType parse_packet();

}
